package vip.ifmm.lehouse.email.service.consumer;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import vip.ifmm.lehouse.common.message.EmailSenderMessage;
import vip.ifmm.lehouse.common.util.RocketMqUtil;
import vip.ifmm.lehouse.email.service.api.EmailSenderService;
import vip.ifmm.lehouse.email.service.strategy.EmailSenderStrategy;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.Arrays;

/**
 * 邮件发送消息消费者
 *
 * @author Fish
 * <p>Email: fishinlove@163.com</p>
 * <p>created by 2019/06/18 19:20:36</p>
 */
@Component
@Slf4j
public class EmailSenderConsumer extends Thread {

    // 消费者主题
    @Value("${consumer-topic}")
    private String consumerTopic = "email";

    // 邮件发送策略
    @Resource
    private EmailSenderStrategy[] senderStrategies = null;

    // 消息队列消费者
    private final DefaultMQPushConsumer consumer = RocketMqUtil.getConsumer(consumerTopic, "*");

    // 初始化工作
    private void init() {
        consumer.setMessageListener((MessageListenerConcurrently) (messageExts, context) -> {
            log.info("接收到发送邮件消息！消息条数：" + messageExts.size());
            for (MessageExt messageExt : messageExts) {
                try {
                    handleMessage(new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET));
                    log.info("消费消息！" + messageExt);
                } catch (UnsupportedEncodingException e) {
                    log.error("消息消费失败！可能是编码不支持！消息是：" + messageExt, e);

                    // 异常重试
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }

            // 消费成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
    }

    @Override
    public void run() {

        log.info("消费者启动！" + consumer);
        init();

        // 启动消费者，接收消息进行处理，发送邮件
        try {
            consumer.start();
        } catch (MQClientException e) {
            log.error("消费者发生异常！", e);
        }
    }

    // 关闭消费者
    public void shutdown() {
        consumer.shutdown();
    }

    // 解析并消费消息
    private void handleMessage(String message) {
        EmailSenderMessage emailSenderMessage = messageFrom(message);
        if (isOverTime(emailSenderMessage)) {
            return; // 过期消息，不进行消费
        }

        //log.info("得到消息：" + emailSenderMessage + "，邮件发送策略：" + Arrays.toString(senderStrategies));
        for (EmailSenderStrategy strategy : senderStrategies) {
            if (strategy.type().equals(emailSenderMessage.getEmailType())) {
                log.info("使用 " + strategy.type() + " 策略发送邮件");
                strategy.sendEmail(emailSenderMessage);
                return;
            }
        }

        log.error("没有对应的邮件发送策略！");
    }

    // 消息是否过期，返回 false 表示可以销费
    private boolean isOverTime(EmailSenderMessage message) {
        return System.currentTimeMillis() > message.getMessageSentTime() + message.getMessageExpiredTime();
    }

    // 解析消息为具体对象，方便操作
    private EmailSenderMessage messageFrom(String message) {
        // 目前就是 JSON 序列化
        return JSON.parseObject(message, EmailSenderMessage.class);
    }
}
